package com.tanjie.blackbean.mqtt;

import com.tanjie.blackbean.mqtt.properties.*;
import com.tanjie.blackbean.mqtt.support.BlackbeanMqttUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@AllArgsConstructor
@Configuration
@EnableConfigurationProperties(value = {BlackbeanMqttProperties.class})
public class BlackbeanMqttAutoConfiguration implements ApplicationContextAware, BeanPostProcessor {

    public static final String ADAPTER_SUFFIX = "-adapter";

    public static final String HANDLER_SUFFIX = "-handler";

    private ConfigurableApplicationContext applicationContext;

    private final BlackbeanMqttProperties properties;

    ApplicationEventPublisher applicationEventPublisher;

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
        properties.getProducers().forEach(this::initProducer);
        properties.getConsumers().forEach(this::initConsumer);
    }

    private MqttPahoClientFactory mqttClientFactory(ChannelConfig config) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(config.getUrl());
        options.setCleanSession(config.isCleanSession());
        options.setUserName(config.getUsername());
        if (config.getPassword() != null) {
            options.setPassword(config.getPassword().toCharArray());
        }
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 初始化生产者，发送mqtt消息
     */
    private void initProducer(ProducerConfig config) {
        String producerName = config.getName();
        String clientId = config.getClientId();
        boolean async = config.isAsync();
        int qos = config.getQos();
        //
        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
        String handlerName = producerName + HANDLER_SUFFIX;
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory(config));
        messageHandler.setAsync(async);
        messageHandler.setAsyncEvents(async);
        messageHandler.setDefaultQos(qos);
        messageHandler.setConverter(new DefaultPahoMessageConverter());
        messageHandler.setApplicationEventPublisher(applicationEventPublisher);
        //
        beanFactory.registerSingleton(handlerName, messageHandler);
        BlackbeanMqttUtil.PRODUCER_MAP.put(producerName, messageHandler);
        log.info(String.format("MQTT: 生产者[%s]已创建", producerName));
    }

    /**
     * 初始化消费者，订阅消息
     */
    private void initConsumer(ConsumerConfig config) {
        int[] qos = config.getQos();
        String[] topics = config.getTopics();
        String consumerName = config.getName();
        String clientId = config.getClientId();
        // spring bean factory
        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
        // 定义消息如何处理的对象，使用线程池处理
        ExecutorChannel executorChannel = new ExecutorChannel(mqttMessageHandleExecutor(config.getThreadPool()));
        executorChannel.setBeanName(consumerName);
        beanFactory.registerSingleton(consumerName, executorChannel);
        // 消息订阅的适配器
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(config), topics);
        DefaultPahoMessageConverter messageConverter = new DefaultPahoMessageConverter();
        // 消息内容设置为byte数组
        messageConverter.setPayloadAsBytes(true);
        adapter.setConverter(messageConverter);
        adapter.setQos(qos);
        adapter.setOutputChannel(executorChannel);
        // 组件会定时执行一些清理或重连 重发操作，需要有定时器，如果不设置，会使用默认的
        adapter.setTaskScheduler(mqttThreadPoolTaskScheduler());
        // qos=1,2时自动响应broker
        adapter.setManualAcks(false);
        // 事件监听传递给spring的event机制
        adapter.setApplicationEventPublisher(applicationEventPublisher);
        beanFactory.registerSingleton(consumerName + ADAPTER_SUFFIX, adapter);
        log.info(String.format("MQTT: 消费者[%s]已创建", consumerName));
    }

    Executor mqttMessageHandleExecutor(ExecutorThreadPoolConfig config) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建时候初始化的线程数
        executor.setCorePoolSize(config.getCorePoolSize());
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(config.getMaxPoolSize());
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(config.getQueueCapacity());
        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setThreadNamePrefix(config.getThreadNamePrefix());
        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        executor.initialize();
        return executor;
    }

    @Bean
    public ThreadPoolTaskScheduler mqttThreadPoolTaskScheduler() {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(20);
        threadPoolTaskScheduler.setThreadNamePrefix("mqtt-maintain-thread");
        return threadPoolTaskScheduler;
    }

}
